/**
 * Copyright(c) 2015 Jade Techonologies Co., Ltd.
 *
 * History:
 *   14-8-1 下午12:44 Created by Daisy
 */
package com.jade.framework.mom.rocketmq;

import java.util.ArrayList;
import java.util.List;

import com.alibaba.rocketmq.client.consumer.DefaultMQPushConsumer;
import com.alibaba.rocketmq.client.consumer.listener.ConsumeConcurrentlyContext;
import com.alibaba.rocketmq.client.consumer.listener.ConsumeConcurrentlyStatus;
import com.alibaba.rocketmq.client.consumer.listener.MessageListenerConcurrently;
import com.alibaba.rocketmq.common.message.MessageExt;
import com.jade.framework.mom.AbstractConsumer;
import com.jade.framework.mom.MomMessage;

/**
 * RocketMq的消费者。
 *
 * @author Daisy
 * @version 1.0 14-8-1 下午12:44
 */
public class RocketmqConsumer
        extends AbstractConsumer
{
    protected DefaultMQPushConsumer consumer;

    public void setConsumer(DefaultMQPushConsumer consumer)
    {
        this.consumer = consumer;
    }

    @Override
    protected void doInit()
    {
        consumer.registerMessageListener(new MessageListenerConcurrently()
        {
            /**
             * 默认msgs里只有一条消息，可以通过设置consumeMessageBatchMaxSize参数来批量接收消息
             */
            public ConsumeConcurrentlyStatus consumeMessage(List<MessageExt> msgs, ConsumeConcurrentlyContext context)
            {
                List<MomMessage> momMessages = new ArrayList<MomMessage>(msgs.size());
                for (MessageExt msgExt : msgs) {
                    momMessages.add(RocketmqUtils.toMomMessage(msgExt));
                }
                boolean result = messageListener.consumeMessage(momMessages);
                return result ? ConsumeConcurrentlyStatus.CONSUME_SUCCESS : ConsumeConcurrentlyStatus.RECONSUME_LATER;
            }
        });
    }

    @Override
    protected void doDestroy()
    {
        //do nothing
    }

    @Override
    protected void doStart()
            throws Exception
    {
        consumer.start();
    }

    @Override
    protected void doStop()
            throws Exception
    {
        consumer.shutdown();
    }
}
